package com.dgw.springbootkafka.sender;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program: springboot-kafka
 * @description: 及时发送消息
 * @author: Mr.Dai
 * @create: 2020-02-22 23:08
 **/
@Component
public class FireProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    private AtomicInteger i = new AtomicInteger(0);

    //@Scheduled(fixedRate = 2000, initialDelay = 5000)
    @Transactional()
    public void send() {
        i.incrementAndGet();
        // 三个参数，topic,key：用户分配partition,value:发送的值
        ProducerRecord<String, String> record = new ProducerRecord<>("test-syn", "topic_" + i.get(), "test-syn-" + i.get());
        // 不关心发送结果
        kafkaTemplate.send(record);
        throw new RuntimeException("fail");
    }

    // @Scheduled(fixedRate = 2000, initialDelay = 5000)
    @Transactional
    public void testExecuteInTransaction() throws InterruptedException {
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
            @Override
            public Object doInOperations(KafkaOperations kafkaOperations) {
                kafkaOperations.send("test-syn", "test executeInTransaction");
                // 模拟代码出现异常
                throw new RuntimeException("fail");
                //return true;
            }
        });
    }
}
